package org.wx.io.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 天生都是boss
 * @program: my-java
 * @description: 线程组
 * @author: Wang.xun
 * @create: 2021-02-06 21:08
 **/
public class SelectorThreadGroup {
    // 线程数组
    SelectorThread[] sts;

    SelectorThreadGroup workGroup = this;
    // 创建监听的渠道
    ServerSocketChannel serverSocketChannel;

    // 用于计数
    AtomicInteger idx = new AtomicInteger(0);

    public SelectorThreadGroup(int num) {
        sts = new SelectorThread[num];
        for (int i = 0; i < num; i++) {
            sts[i] = new SelectorThread(this);
            // 开启每个线程
            new Thread(sts[i]).start();
        }
    }

    /**
     * 设置boss的work组
     * 如果不设置，那么boss组 执行监听和通讯任务
     * 设置了work组，那么boss组只进行监听，work组进行通讯
     * @param workGroup 工作组
     */
    public void setWorker(SelectorThreadGroup workGroup){
        this.workGroup = workGroup;
    }

    /**
     * 绑定端口
     * @param address
     */
    public void bind(InetSocketAddress address){
        try {
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(address);
        } catch (IOException e) {
            e.printStackTrace();
        }
        // 将渠道注册到某个selector中
        nextSelector(serverSocketChannel);
    }

    public void bind(int port){
        bind(new InetSocketAddress("192.168.106.1",port));
    }


    /**
     * 通道选择selector
     * @param channel
     */
    public void nextSelector(Channel channel) {
        try {
            // 如果是ServerSocketChannel 那么就在boss线程中取一个selector进行注册
            SelectorThread selector;
            if(channel instanceof ServerSocketChannel){
                selector = next();
                selector.linkedBlockingQueue.put(channel);
                selector.setWorker(workGroup);
                // 唤醒 selector 让 select() 方法立即返回
            }else{ // SocketChannel
                selector = workGroupNext();
                selector.linkedBlockingQueue.put(channel);
            }
            selector.selector.wakeup();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 取本线程组中的下一个
     */
    private SelectorThread next() {
        int id = idx.incrementAndGet() % sts.length;
        return sts[id];
    }

    /**
     * 在工作线程组中取下一个
     * @return
     */
    private SelectorThread workGroupNext() {
        int id = idx.incrementAndGet() % workGroup.sts.length;
        return workGroup.sts[id];
    }

}
